package com.wudl.flink.stream.tablesql

import com.wudl.flink.stream.source.StationLog
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.shaded.jackson2.org.yaml.snakeyaml.events.StreamEndEvent
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.sources.CsvTableSource
import org.apache.flink.types.Row

object CreateTableStream {
  def main(args: Array[String]): Unit = {
    /**
     * 初始化上下文的环境
     */
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().build()
    val tableEnvironment: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, settings)
    //两个隐式转换
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.scala._

    //  读取数据源

    //    val stream = streamEnv.readTextFile(getClass.getResource("/station.log").getPath)

    val stream: DataStream[StationLog] = streamEnv.socketTextStream("10.204.125.109", 8888)
      .map(line => {
        val arr = line.split(",")

//        println(arr)
        //        new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
        new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
      })
    tableEnvironment.registerDataStream("t_table2", stream)
    //
    //    tableEnvironment.sqlQuery("select * from  t_table2")
    //    val table: Table = tableEnvironment.scan("t_table2")
    //
    //    table.printSchema()

    // 修改字段名称
//    val table: Table = tableEnvironment.fromDataStream(stream, 'id, 'call_out, 'call_in)
//    table.printSchema()

    //查询过滤
    //    val table: Table = tableEnvironment.fromDataStream(stream)
    //    val result: Table = table.filter('callType == "success")
    //    val value = tableEnvironment.toAppendStream[Row](result)
    //    value.print()

    val table: Table = tableEnvironment.fromDataStream(stream)


    /** *****************************/
    // filter 的过滤
//    var result:Table=table.filter('callType==="success").select("sid")
//    val ds:DataStream[Row] = tableEnvironment.toAppendStream[Row](result)
//    ds.print()

    //  分组聚合
   val result:Table =  table.groupBy('sid).select('sid,'sid.count as 'log_count)
    tableEnvironment.toRetractStream[Row](result).filter(_._1==true).print()

    streamEnv.execute()
  }

}
